package arp.repository.springdatamongodb;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.locks.LockSupport;

import org.bson.Document;
import org.springframework.data.domain.Sort;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;

import arp.enhance.ProcessInfo;
import arp.process.ProcessContext;
import arp.process.ThreadBoundProcessContextArray;
import arp.repository.PersistenceRepository;

public abstract class MappedMongodbRepository<E, DTO, I> extends
		PersistenceRepository<E, I> {

	protected MongoTemplate mongoTemplate;

	int lockRetryCount = 300;

	private CurrentTimeMillisClock clock = CurrentTimeMillisClock.getInstance();

	protected long maxLockTime;

	protected MappedMongodbRepository(MongoTemplate mongoTemplate) {
		this(mongoTemplate, Long.MAX_VALUE);
	}

	protected MappedMongodbRepository(MongoTemplate mongoTemplate,
			long maxLockTime) {
		if (mongoTemplate == null) {
			initAsMock();
		} else {
			this.mongoTemplate = mongoTemplate;
			this.maxLockTime = maxLockTime;
		}
	}

	protected abstract E toEntity(DTO dto);

	protected abstract Class<DTO> getDtoClass();

	protected abstract DTO toDTO(E entity);

	@Override
	protected E findByIdForUpdateImpl(I id) {
		boolean hasLock = acquireLock(id);
		if (!hasLock) {
			return null;
		}
		return findByIdImpl(id);
	}

	private boolean createLock(I id) {
		Class<DTO> dtoCls = getDtoClass();
		String collectionName = "arp_repo_state_" + dtoCls.getSimpleName();
		Map<String, Object> lock = new HashMap<>();
		lock.put("_id", id);
		lock.put("state", 1);
		lock.put("time", clock.now());
		ProcessContext processContext = ThreadBoundProcessContextArray
				.getProcessContext();
		lock.put("piid", processContext.getProcessInfoId());
		try {
			mongoTemplate.insert(lock, collectionName);
			return true;
		} catch (Exception e) {
			return false;
		}
	}

	// 如果没有数据也返回，返回false，有数据的话那就争夺锁，抢到了返回true，抢不到抛出CanNotAcquireLockException
	private boolean acquireLock(I id) {
		ProcessContext processContext = ThreadBoundProcessContextArray
				.getProcessContext();
		Class<DTO> dtoCls = getDtoClass();
		String collectionName = "arp_repo_state_" + dtoCls.getSimpleName();
		Document cmd = new Document();
		cmd.put("findAndModify", collectionName);
		Document query = new Document();
		query.put("_id", id);
		List<Document> or = new ArrayList<>();
		Document state = new Document();
		state.put("state", 0);
		Document time = new Document();
		long now = clock.now();
		long unlockTime = now - maxLockTime;
		Document lt = new Document();
		lt.put("$lt", unlockTime);
		time.put("time", lt);
		or.add(state);
		or.add(time);
		query.put("$or", or);
		cmd.put("query", query);
		Document update = new Document();
		Document set = new Document();
		set.put("state", 1);
		set.put("piid", processContext.getProcessInfoId());
		set.put("time", now);
		update.put("$set", set);
		cmd.put("update", update);
		cmd.put("new", false);

		int counter = lockRetryCount;
		do {
			Document doc = mongoTemplate.executeCommand(cmd);
			Document lastErrorObject = doc.get("lastErrorObject",
					Document.class);
			boolean updatedExisting = lastErrorObject
					.getBoolean("updatedExisting");
			if (updatedExisting) {
				return true;
			}
			Query lockExistsCheckQuery = new Query();
			lockExistsCheckQuery.addCriteria(Criteria.where("_id").is(id));
			boolean lockExists = mongoTemplate.exists(lockExistsCheckQuery,
					collectionName);
			if (!lockExists) {
				Query entityExistsCheckQuery = new Query();
				entityExistsCheckQuery
						.addCriteria(Criteria.where("_id").is(id));
				boolean entityExists = mongoTemplate.exists(
						entityExistsCheckQuery, dtoCls);
				if (!entityExists) {
					return false;
				} else {
					if (createLock(id)) {
						return true;
					} else {
						continue;
					}
				}
			}

			if (counter > 200) {
				--counter;
			} else if (counter > 100) {
				--counter;
				Thread.yield();
			} else if (counter > 0) {
				--counter;
				LockSupport.parkNanos(1L);
			} else {
				Document lockQueryCmd = new Document();
				lockQueryCmd.put("find", collectionName);
				Document filter = new Document();
				filter.put("_id", id);
				lockQueryCmd.put("filter", filter);
				Document lockQueryDoc = mongoTemplate
						.executeCommand(lockQueryCmd);
				Document cursor = lockQueryDoc.get("cursor", Document.class);
				List<Document> firstBatch = cursor.getList("firstBatch",
						Document.class);
				Document lockDoc = firstBatch.get(0);
				int piid = lockDoc.getInteger("piid");
				ProcessInfo processInfoGotLock = ProcessContext
						.getProcessInfo(piid);
				String processDesc;
				if (!processInfoGotLock.getProcessName().trim().isEmpty()) {
					processDesc = processInfoGotLock.getProcessName();
				} else {
					processDesc = processInfoGotLock.getClsName() + "."
							+ processInfoGotLock.getMthName();
				}
				throw new CanNotAcquireLockException(dtoCls.getName(),
						processDesc);
			}
		} while (true);
	}

	@Override
	protected E findByIdImpl(I id) {
		DTO dto = mongoTemplate.findById(id, getDtoClass());
		if (dto == null) {
			return null;
		}
		return toEntity(dto);
	}

	@Override
	protected E saveIfAbsentImpl(I id, E entity) {
		DTO dto = toDTO(entity);
		if (!createLock(id)) {
			acquireLock(id);
			return findByIdImpl(id);
		}
		mongoTemplate.insert(dto);
		return null;
	}

	@Override
	protected void updateBatchImpl(Map<I, E> entitiesToUpdate) {
		entitiesToUpdate.forEach((id, ett) -> {
			updateImpl(id, ett);
		});
	}

	@Override
	protected void updateImpl(I id, E entity) {
		mongoTemplate.save(toDTO(entity));
	}

	private void unlock(I id) {
		Query query = new Query();
		query.addCriteria(Criteria.where("_id").is(id));
		Update update = new Update();
		update.set("state", 0);
		Class<DTO> dtoCls = getDtoClass();
		String collectionName = "arp_repo_state_" + dtoCls.getSimpleName();
		mongoTemplate.updateFirst(query, update, collectionName);
	}

	@Override
	protected void saveBatchImpl(Map<I, E> entities) {
		List<DTO> dtos = new ArrayList<>();
		entities.values().forEach((ett) -> {
			dtos.add(toDTO(ett));
		});
		mongoTemplate.insert(dtos, getDtoClass());
	}

	@Override
	protected void saveImpl(I id, E entity) {
		mongoTemplate.save(toDTO(entity));
	}

	@Override
	protected void removeBatchImpl(Set<I> ids) {
		ids.forEach((id) -> {
			removeImpl(id);
		});
	}

	@Override
	protected void removeImpl(I id) {
		DTO dto = mongoTemplate.findById(id, getDtoClass());
		if (dto != null) {
			mongoTemplate.remove(dto);
			removeLock(id);
		}
	}

	private void removeLock(I id) {
		Query query = new Query();
		query.addCriteria(Criteria.where("_id").is(id));
		Class<DTO> dtoCls = getDtoClass();
		String collectionName = "arp_repo_state_" + dtoCls.getSimpleName();
		mongoTemplate.remove(query, collectionName);
	}

	@Override
	protected void unlockBatchImpl(Set<I> ids) {
		ids.forEach((id) -> {
			unlockImpl(id);
		});
	}

	@Override
	protected void unlockImpl(I id) {
		unlock(id);
	}

	public I findMaxId() {
		if (mongoTemplate == null) {
			return null;
		}
		Query query = new Query();
		query.with(Sort.by(Sort.Direction.DESC, "_id"));
		query.limit(1);
		DTO maxDto = mongoTemplate.findOne(query, getDtoClass());
		if (maxDto == null) {
			return null;
		}
		return getId(toEntity(maxDto));
	}

	public List<E> findAllByField(String fieldName, Object fieldValue) {
		Query query = new Query();
		query.addCriteria(Criteria.where(fieldName).is(fieldValue));
		List<DTO> dtos = mongoTemplate.find(query, getDtoClass());
		List<E> etts = new ArrayList<>();
		for (DTO dto : dtos) {
			etts.add(toEntity(dto));
		}
		return etts;
	}
}
